package com.workcell.workdata.xthings.kafka.listener;

import com.workcell.workdata.xthings.constant.KafKaTopicConstant;
import com.workcell.workdata.xthings.kafka.content.MqContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * @Author 廖博文
 * @Date 2023/5/25 11:11
 * @Description 监听收到的消息
 */
@Slf4j
@Component
public class KafkaListenerService {

	@Resource
	private MqContext mqContext;

	@KafkaListener(topics = {KafKaTopicConstant.DEMO_BO}, errorHandler = "kafkaListenerResultHandler")
	public void consumptionListener(ConsumerRecord<Object, Objects> consumerRecord, Acknowledgment ack){
		try {
			// 策略模式
			mqContext.getMqService(consumerRecord.topic()).handlerKafkaMessage(consumerRecord);
			// 手动确认
			ack.acknowledge();
		} catch (Exception e){
			log.error("Kafka消息队列消费日志，消费前异常：{}", e.getMessage());
		}
	}

}
